package com.knight.hadoop.day12.zuoye;

import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
import java.util.List;

import org.apache.commons.beanutils.BeanUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import com.knight.hadoop.day10.preprocess.WebLogObject;

public class Pageviews {
	static NullWritable v = NullWritable.get();

	static class PageviewsMapper extends Mapper<LongWritable, Text, LogBean2, LogBean2> {

		static LogBean2 logbean = new LogBean2();
		static Text t = new Text();

		@Override
		protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
			String line = value.toString();
			String[] splitStr = line.split("\001");
			String date = splitStr[1];
			String ip = splitStr[2];
			String sessionId = splitStr[3];
			String requestURL = splitStr[4];
			String referal = splitStr[5];
			int stepNum = Integer.parseInt(splitStr[7]);
			long stayTime = Long.parseLong(splitStr[6]);

			//logbean.set(date, ip, sessionId, requestURL, referal, stayTime, stepNum);

			//t.set(sessionId);

			context.write(logbean, logbean);
		}
	}

	static class PageviewsReducer extends Reducer<LogBean2, LogBean2, Text, NullWritable> {

		static Text k = new Text();

		@Override
		protected void reduce(LogBean2 key, Iterable<LogBean2> values, Context context)
				throws IOException, InterruptedException {

			List<LogBean2> list = new ArrayList<>();

			for (LogBean2 bean : values) {
				LogBean2 logBean = new LogBean2();
				try {
					BeanUtils.copyProperties(logBean, bean);
				} catch (IllegalAccessException e) {
					e.printStackTrace();
				} catch (InvocationTargetException e) {
					e.printStackTrace();
				}
				list.add(logBean);
			}
			
			int last = list.size() - 1;
			
			StringBuilder sb = new StringBuilder();

			sb.append(list.get(0).getSessionId() + "\001");
			sb.append(list.get(0).getDate() + "\001");
			sb.append(list.get(last).getDate() + "\001");
			sb.append(list.get(0).getRequestURL() + "\001");
			sb.append(list.get(last).getRequestURL() + "\001");
			sb.append(list.size() + "\001");
			sb.append(list.get(0).getIp() + "\001");
			sb.append(list.get(0).getReferal());

			k.set(sb.toString());
			context.write(k, v);
		}
	}

	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();
		Job job = Job.getInstance(conf);
		job.setJarByClass(WebLogObject.class);
		job.setMapperClass(PageviewsMapper.class);
		job.setReducerClass(PageviewsReducer.class);
		job.setGroupingComparatorClass(LogPageViewsComparator.class);
		
		
		job.setMapOutputKeyClass(LogBean2.class);
		job.setMapOutputValueClass(LogBean2.class);

		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(NullWritable.class);

		FileInputFormat.setInputPaths(job, new Path("E:\\home\\webLog\\step2\\output"));
		FileOutputFormat.setOutputPath(job, new Path("E:\\home\\webLog\\step3\\output"));

		boolean res = job.waitForCompletion(true);

		System.exit(res ? 0 : 1);
	}
}
